package com.econ.powercloud.ems.service.biz;

import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import com.econ.powercloud.clickhouse.entity.EmsFaultData;
import com.econ.powercloud.common.base.facade.enums.StatusType;
import com.econ.powercloud.common.base.facade.exception.CommonRuntimeException;
import com.econ.powercloud.common.base.facade.pojo.PageResult;
import com.econ.powercloud.common.base.facade.util.JacksonUtil;
import com.econ.powercloud.ems.common.enums.*;
import com.econ.powercloud.ems.common.pojo.CacheOperationVO;
import com.econ.powercloud.ems.common.pojo.EmsAnimateData;
import com.econ.powercloud.ems.common.pojo.EmsKafkaYxDataDTO;
import com.econ.powercloud.ems.dal.entity.*;
import com.econ.powercloud.ems.pojo.*;
import com.econ.powercloud.ems.pojo.requestVO.EmsFaultRequestVO;
import com.econ.powercloud.ems.pojo.scene.HisDataPidsVO;
import com.econ.powercloud.ems.pojo.scene.HisDataRequestVO;
import com.econ.powercloud.ems.pojo.scene.OpenApiHisDataVO;
import com.econ.powercloud.ems.service.common.QueryHisDataCommonService;
import com.econ.powercloud.ems.service.dao.*;
import com.econ.powercloud.ems.util.RedisCacheUtil;
import com.econ.powercloud.ems.util.SendKafkaDateUtil;
import com.econ.powercloud.ems.util.TimeUtil;
import com.econ.powercloud.powernode.gateway.common.domain.UserDTO;
import com.google.common.collect.Lists;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections.MapUtils;
import org.springframework.beans.PropertyValues;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.util.Assert;
import org.springframework.util.CollectionUtils;

import java.util.*;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.stream.Collectors;


/**
 * @author zjr
 * @create 2021/7/30 上午8:58
 * @description
 */
@Service
@Slf4j
public class FaultService {

    @Autowired
    private CompanyUtilService companyUtilService;
    @Autowired
    private EmsEnergyStationDAO emsEnergyStationDAO;
    @Autowired
    private EmsEnergyEquipmentDAO emsEnergyEquipmentDAO;
    @Autowired
    private EmsDeviceDAO emsDeviceDAO;
    @Autowired
    private IotProductYxDAO iotProductYxDAO;
    @Autowired
    private IotProductYcDAO iotProductYcDAO;
    @Autowired
    private QueryHisDataCommonService queryHisDataCommonService;
    @Autowired
    private AuthUserDAO authUserDAO;
    @Autowired
    private ClickHouseBizService clickHouseBizService;
    @Autowired
    private EmsFaultDAO emsFaultDAO;
    @Autowired
    private RedisCacheUtil redisCacheUtil;
    @Autowired
    private SendKafkaDateUtil sendKafkaDateUtil;

    /**
     * 获取故障列表
     *
     * @param user
     * @param requestVO
     * @return
     */
    public PageResult<List<EmsFaultVO>> findFaultListByConditionFromClickHouse(UserDTO user, EmsFaultRequestVO requestVO, int pageNo, int pageSize){
        List<String> companyIds = companyUtilService.getCompanyIdsByRole(user);
        List<EmsEnergyStation> emsEnergyStations = emsEnergyStationDAO.selectListByConpanyList(companyIds);
        List<String> energyStationIdList= Lists.newArrayList();
        if (!CollectionUtils.isEmpty(emsEnergyStations)){
            energyStationIdList.addAll(emsEnergyStations.stream().map(EmsEnergyStation::getEnergyStationId).collect(Collectors.toList()));
        }
        Page<EmsFaultData> page = clickHouseBizService.findListByCondition(requestVO, energyStationIdList, pageNo, pageSize);
        List<EmsFaultData> list = page.getRecords();
        if (!CollectionUtils.isEmpty(list)){
            List<EmsFaultVO> faultVOList = JacksonUtil.convertToList(list, EmsFaultVO.class);
            //补充VO中的数据
            this.assignEmsFaultVO(faultVOList);
            return PageResult.<List<EmsFaultVO>>builder().data(faultVOList).pageNo((int)page.getCurrent()).totalCount((int)page.getTotal()).build();
        }
        return PageResult.<List<EmsFaultVO>>builder().data(Collections.EMPTY_LIST).pageNo((int)page.getCurrent()).totalCount((int)page.getTotal()).build();
    }

    /**
     * 查询单个故障详情
     *
     * @param id
     * @return
     */
    public EmsFaultVO selectById(String id) {
        EmsFaultData emsFaultData = clickHouseBizService.findEmsFaultById(id);
        Assert.notNull(emsFaultData, FaultDesc.FAILURE_NOT_FOUNT.getDesc());
        EmsFaultVO emsFaultVO = JacksonUtil.convertToObj(emsFaultData, EmsFaultVO.class);
        this.assignEmsFaultVO(Collections.singletonList(emsFaultVO));
        return emsFaultVO;
    }

    /**
     * 处理单个故障号
     */
    public String modifyFaultById(UserDTO user, String id) {
        EmsFaultData emsFault = clickHouseBizService.findEmsFaultById(id);
        if (!EventDealStatusType.UN_DEAL.getCode().equals(emsFault.getDealState())) {
            throw new CommonRuntimeException(StatusType.FAIL, "该故障号不处于未处理状态!");
        }
        emsFault.setDealState(EventDealStatusType.DEAL.getCode());
        emsFault.setGmtModify(System.currentTimeMillis());
        emsFault.setOperatorId(user.getId());
        clickHouseBizService.updateEmsFaultById(emsFault);
        EmsFault emsFaultRedis = redisCacheUtil.selectLeastFaultData(emsFault.getDeviceCode(), emsFault.getFaultPid(), emsFault.getFaultResource());
        if (emsFaultRedis.getGmtCreate() <= emsFault.getGmtCreate()) {
            //更新redis
            EmsFaultRedisVO emsFaultRedisVO = EmsFaultRedisVO.builder().faultResource(emsFault.getFaultResource())
                    .faultPid(emsFault.getFaultPid()).gmtCreate(emsFault.getGmtCreate()).faultVal(emsFault.getFaultVal())
                    .deviceCode(emsFault.getDeviceCode()).dealState(emsFault.getDealState()).build();
            redisCacheUtil.addEmsFaultList(emsFault.getDeviceCode(), Lists.newArrayList(emsFaultRedisVO));
        }
        //判断是否是最新的故障告警数据被处理，是则推送动画状态，灯停止闪烁
        List<String> pids = new ArrayList<>();
        pids.add(emsFault.getFaultPid());
        List<EmsFaultData> emsFaultDataList = clickHouseBizService.findFaultListByPid(pids,emsFault.getFaultResource());
        if (emsFault.getGmtCreate().compareTo(emsFaultDataList.get(0).getGmtCreate())==0){
            //推送动画状态数据至kafka
            EmsAnimateData emsAnimateData = EmsAnimateData.builder()
                    .animate(AnimateType.NORMAL.getCode())
                    .id(id)
                    .pid(emsFault.getFaultPid())
                    .deviceCode(emsFault.getDeviceCode())
                    .val(emsFault.getFaultVal())
                    .build();
            List<CacheOperationVO> sendData = sendKafkaDateUtil.getSendData(OperationTypeEnum.DELETE, Lists.newArrayList(emsAnimateData), CacheTopicType.ANIMATE);
            sendKafkaDateUtil.sendCommonOperationData(sendData);
        }
        return id;
    }

    /**
     * 处理多个故障号
     */
    public List<String> modifyFaultByIds(UserDTO user, EmsFaultRequestVO requestVO) {
        List<EmsFaultData> emsFaultList = clickHouseBizService.findListByTime(requestVO)
                .stream().filter(d->EventDealStatusType.UN_DEAL.getCode().equals(d.getDealState())).collect(Collectors.toList());
        Assert.notEmpty(emsFaultList,"所有故障告警都已处理");
        List<String> pids = new ArrayList<>();
        ExecutorService executor = Executors.newFixedThreadPool(5);
        emsFaultList.forEach(d->{
            d.setDealState(EventDealStatusType.DEAL.getCode());
            d.setGmtModify(System.currentTimeMillis());
            d.setOperatorId(user.getId());
            executor.execute(()->{
                EmsFault emsFaultRedis = redisCacheUtil.selectLeastFaultData(d.getDeviceCode(), d.getFaultPid(), d.getFaultResource());
                if (emsFaultRedis.getGmtCreate() <= d.getGmtCreate()) {
                    EmsFaultRedisVO emsFaultRedisVO = EmsFaultRedisVO.builder().faultResource(d.getFaultResource())
                            .faultPid(d.getFaultPid()).gmtCreate(d.getGmtCreate()).faultVal(d.getFaultVal())
                            .deviceCode(d.getDeviceCode()).dealState(d.getDealState()).build();
                    redisCacheUtil.addEmsFaultList(d.getDeviceCode(), Lists.newArrayList(emsFaultRedisVO));
                }
                log.info(Thread.currentThread().getName()+"故障告警处理成功pid:{}",d.getFaultPid());
            });
            pids.add(d.getFaultPid());
        });
        Long start = System.currentTimeMillis();
        clickHouseBizService.updateDealStateAndGmtModifyOperatorId(EventDealStatusType.DEAL.getCode(), user.getId(), System.currentTimeMillis(), requestVO.getStartTime(), requestVO.getEndTime());
        log.info("ck更新耗时: {}, size: {}", (System.currentTimeMillis() - start), emsFaultList.size());
        //判断是否是最新的故障告警数据被处理，是则推送动画状态，灯停止闪烁
        List<Object> kafkaSendDataList = new ArrayList<>();
        List<EmsFaultData> allEmsFaultDataList = clickHouseBizService.findFaultListByPid(pids,emsFaultList.get(0).getFaultResource());
        Map<String,List<EmsFaultData>> emsFaultDataMap = allEmsFaultDataList.stream().collect(Collectors.groupingBy(EmsFaultData::getFaultPid));
        emsFaultList.forEach(f->{
            List<EmsFaultData> emsFaultDataList = emsFaultDataMap.get(f.getFaultPid());
            if (!emsFaultDataList.isEmpty()&&f.getGmtCreate().compareTo(emsFaultDataList.get(0).getGmtCreate())==0){
                //推送动画状态数据至kafka
                EmsAnimateData emsAnimateData = EmsAnimateData.builder()
                        .animate(AnimateType.NORMAL.getCode())
                        .id(f.getId())
                        .pid(f.getFaultPid())
                        .deviceCode(f.getDeviceCode())
                        .val(f.getFaultVal())
                        .build();
                kafkaSendDataList.add(emsAnimateData);
            }
        });
        List<CacheOperationVO> sendData = sendKafkaDateUtil.getSendData(OperationTypeEnum.DELETE, kafkaSendDataList, CacheTopicType.ANIMATE);
        sendKafkaDateUtil.sendCommonOperationData(sendData);
        return pids;
    }


    /**
     * 对EmsFaultVO中的数据进行赋值的操作
     */
    private void assignEmsFaultVO(List<EmsFaultVO> list) {
        Map<String, String> stationMap = emsEnergyStationDAO.selectAll().stream().collect(Collectors.toMap(EmsEnergyStation::getEnergyStationId, EmsEnergyStation::getEnergyStationName));
        Map<String, String> equipmentMap = emsEnergyEquipmentDAO.selectAll().stream().collect(Collectors.toMap(EmsEnergyEquipment::getEnergyEquipmentId, EmsEnergyEquipment::getEnergyEquipmentName));
        //查询设备名称
        List<String> deviceIds = list.stream().map(EmsFaultVO::getDeviceCode).collect(Collectors.toList());
        Map<String, String> deviceMap = emsDeviceDAO.selectByDeviceCodes(deviceIds).stream().collect(Collectors.toMap(EmsDevice::getDeviceCode, EmsDevice::getDeviceName));
        //获取用户信息
        List<AuthUser> authUserList = authUserDAO.list();
        Map<String, String> userMap = authUserList.stream().collect(Collectors.toMap(AuthUser::getId, AuthUser::getName));
        list.stream().forEach(d -> {
            d.setDeviceName(deviceMap.get(d.getDeviceCode()));
            d.setEnergyStationName(stationMap.get(d.getEnergyStationId()));
            d.setEnergyEquipmentName(equipmentMap.get(d.getEnergyEquipmentId()));
            d.setOperatorName(d.getOperatorId() == null ? null : userMap.get(d.getOperatorId()));
            d.setFaultResourceName(EventSourceType.getDescFromCode(d.getFaultResource()));
            d.setDealStateName(EventDealStatusType.getDescFromCode(d.getDealState()==null?0:d.getDealState()));
            d.setFaultTypeName(d.getFaultType() == null?null : AlarmOrFaultState.getDescFromCode(d.getFaultType()));
            //为了clickhouse的性能问题,在插入数据时,会把处理时间设置成0
            if(Objects.equals(d.getGmtModify(),0L)){
                d.setGmtModify(null);
            }
        });
    }

    /**
     * @Description: 获取平台事件对应点号历史数据
     * @Param: [faultPid, startTime, endTime, pageNo, pageSize]
     * @Return:
     */
    public PageResult<List<EmsFaultDitHisVO>> getFaultDitHis(String id, Long startTime, Long endTime, int pageNo, int pageSize) {
        startTime = TimeUtil.getDayStartTime(new Date(startTime));
        endTime = TimeUtil.getDayEndTime(new Date(endTime));
        EmsFaultData emsFault = clickHouseBizService.findEmsFaultById(id);
        Assert.notNull(emsFault, "故障不存在");
        EmsDevice emsDevice = emsDeviceDAO.selectByDeviceCode(emsFault.getDeviceCode());
        Assert.notNull(emsDevice, "设备不存在");
        List<HisDataPidsVO> pidsVOList = Lists.newArrayList();
        List<String> dits = Lists.newArrayList();
        dits.add(emsFault.getRealPid().replace(emsDevice.getInitDeviceCode() + ".", ""));
        HisDataPidsVO pidsVO = HisDataPidsVO.builder().deviceCode(emsDevice.getInitDeviceCode()).dits(dits).build();
        pidsVOList.add(pidsVO);
        HisDataRequestVO hisAnologRequestVO = HisDataRequestVO.builder().startTime(startTime)
                .endTime(endTime).pids(pidsVOList).build();
        Map<String,List<OpenApiHisDataVO>> hisDataVOList;
        if (MeasureType.yc.getCode().equals(emsFault.getRealMeasureType())) {
            //获取遥测历史数据
            hisDataVOList = queryHisDataCommonService.getYcDataByListPid(hisAnologRequestVO);
        } else {
            //获取历史遥信历史数据
            hisDataVOList = queryHisDataCommonService.getYxDataByListPid(hisAnologRequestVO);
        }
        if (MapUtils.isEmpty(hisDataVOList)) {
            return PageResult.<List<EmsFaultDitHisVO>>builder().totalCount(0).pageNo(pageNo).data(null).build();
        }
        List<EmsFaultDitHisVO> hisVOS = Lists.newArrayList();
        hisDataVOList.forEach((key, value) ->
                hisVOS.addAll(value.stream().map(d -> new EmsFaultDitHisVO(d.getV(), d.getTime())).collect(Collectors.toList())));
        //按时间倒序排序
        List<EmsFaultDitHisVO> result = hisVOS.stream().filter(Objects::nonNull).sorted(Comparator.comparing(EmsFaultDitHisVO::getTime).reversed()).collect(Collectors.toList());
        int size = result.size();
        if (MeasureType.yx.getCode().equals(emsFault.getRealMeasureType()) && (!CollectionUtils.isEmpty(result))) {
            hisVOS.clear();
            hisVOS.add(result.get(0));
            for (int i = 1; i < size; i++) {
                //遥信数据去重
                if (!result.get(i).getVal().equals(result.get(i - 1).getVal())) {
                    hisVOS.add(result.get(i));
                }
            }
            size = hisVOS.size();
            if (pageNo * pageSize > size) {
                result = hisVOS.subList((pageNo - 1) * pageSize, size);
            } else {
                result = hisVOS.subList((pageNo - 1) * pageSize, pageNo * pageSize);
            }
        }
        return PageResult.<List<EmsFaultDitHisVO>>builder().totalCount(size)
                .pageNo(pageNo)
                .data(result).build();
    }

    /**
     * @Description: 根据故障号获取点号id以及点号名称
     * @Param: [id]
     * @Return:
     */
    public ApiDataVO getFaultDitAndName(String id) {
//        EmsFault fault = emsFaultDAO.getById(id);
        EmsFaultData fault = clickHouseBizService.findEmsFaultById(id);
        Assert.notNull(fault, "故障不存在");
        EmsDevice device = emsDeviceDAO.selectByDeviceCode(fault.getDeviceCode());
        Assert.notNull(device, "设备不存在");
        String dit = fault.getRealPid().replace(device.getInitDeviceCode() + ".", "");
        ApiDataVO result=new ApiDataVO();
        if(MeasureType.yx.getCode().equals(fault.getRealMeasureType())){
            IotProductYx yx = iotProductYxDAO.selectYxByDit(dit, device.getProductKey());
            result.setCode(yx == null?null : yx.getDit());
            result.setName(yx == null?null : yx.getName());
        }else {
            IotProductYc yc = iotProductYcDAO.selectYcByDit(dit, device.getProductKey());
            result.setCode(yc == null?null : yc.getDit());
            result.setName(yc == null?null : yc.getName());
        }
        return result;
    }

    public Boolean syncList() {
        List<EmsFault> list = emsFaultDAO.list();
        clickHouseBizService.emsFaultListDataIntoClickHouse(list);
        return true;
    }

    public void emsFaultToRedisAndClickHouse(EmsFault emsFault) {
        if (emsFault == null) {
            return;
        }
        EmsFaultRedisVO emsFaultRedisVOBuilder = EmsFaultRedisVO.builder().faultPid(emsFault.getFaultPid()).faultVal(emsFault.getFaultVal())
                .deviceCode(emsFault.getDeviceCode()).faultResource(emsFault.getFaultResource())
                .dealState(emsFault.getDealState()).gmtCreate(emsFault.getGmtCreate()).build();
        redisCacheUtil.addEmsFaultList(emsFault.getDeviceCode(), Lists.newArrayList(emsFaultRedisVOBuilder));
        clickHouseBizService.emsFaultDataIntoClickHouse(emsFault);
    }

    /**
     * 获取故障告警的最新的一条记录
     * @param deviceCode
     * @param pid
     * @param eventSourceType
     * @return
     */
    public EmsFault selectLeastFaultData(String deviceCode, String pid, Integer eventSourceType){
        EmsFault redisResult = redisCacheUtil.selectLeastFaultData(deviceCode, pid, eventSourceType);
        if(redisResult != null){
            return redisResult;
        }
        //如果没有查询到就从ClickHouse中获取
        EmsFault clickHouseResult = clickHouseBizService.selectOneByFaultPidAndFaultResource(pid,eventSourceType);
        if(clickHouseResult != null){
            EmsFaultRedisVO faultRedisVO = JacksonUtil.convertToObj(clickHouseResult,EmsFaultRedisVO.class);
            redisCacheUtil.addEmsFaultList(deviceCode,Lists.newArrayList(faultRedisVO));
        }
        return clickHouseResult;
    }

    public Map<String,EmsFault> selectLeastFaultList(List<EmsKafkaYxDataDTO> yxList){
        List<EmsFault> faultList = redisCacheUtil.selectLeastFaultByKafkaYxData(yxList);
        Map<String,EmsFault> result = faultList.stream().collect(Collectors.toMap(EmsFault::getFaultPid,d->d,(p1,p2)->p1));
        //更新redis中的最新数据
        List<EmsFaultRedisVO> redisVOList = Lists.newArrayList();
        yxList.stream().forEach(d->{
            if(result.get(d.getPid()) == null){
                //如果没有查询到就从ClickHouse中获取
                EmsFault clickHouseResult = clickHouseBizService.selectOneByFaultPidAndFaultResource(d.getPid(),EventSourceType.YX.getCode());
                if(clickHouseResult != null){
                    EmsFaultRedisVO faultRedisVO = JacksonUtil.convertToObj(clickHouseResult,EmsFaultRedisVO.class);
                    redisVOList.add(faultRedisVO);
                    result.put(d.getPid(),clickHouseResult);
                }
            }
        });
        if(!CollectionUtils.isEmpty(redisVOList)){
            redisVOList.stream().collect(Collectors.groupingBy(d->d.getDeviceCode())).forEach((k,v)->{
                redisCacheUtil.addEmsFaultList(k, v);
            });
        }
        return result;
    }
}
